package com.ht.api.flink;

import java.util.Objects;
import java.util.Properties;

//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.api.common.time.Time;
//import org.apache.flink.connector.base.DeliveryGuarantee;
//import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
//import org.apache.flink.connector.kafka.sink.KafkaSink;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class MysqlToKafka {
//
//    @Data
//    public static class Config {
//        // MySQL配置
//        private String hostname = "127.0.0.1";
//        private String username = "root";
//        private String password = "aixi_2028";
//        private int port = 3306;
//        private String database = "test";
//        private String table = "users";
//
//        // Kafka配置
//        private String bootstrapServers = "192.168.0.240:9092";
//        private String topic = "mysql_data";
//    }
//
//    public static void main(String[] args) throws Exception {
//        Config config = new Config();
//        
//        // 1. 设置环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        
//        // 设置检查点
//        env.enableCheckpointing(3000);
//        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        checkpointConfig.setMinPauseBetweenCheckpoints(500);
//        checkpointConfig.setCheckpointTimeout(60000);
//        checkpointConfig.setMaxConcurrentCheckpoints(1);
//        checkpointConfig.setExternalizedCheckpointCleanup(
//            CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        
//        // 设置重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
//        
//        // 2. MySQL Source
//     // 2. MySQL Source
//        Properties jdbcProperties = new Properties();
//        jdbcProperties.setProperty("connectTimeout", "30000");
//        jdbcProperties.setProperty("socketTimeout", "30000");
//        jdbcProperties.setProperty("useSSL", "false");
//
//        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
//            .hostname(config.getHostname())
//            .port(config.getPort())
//            .databaseList(config.getDatabase())
//            .tableList(config.getDatabase() + "." + config.getTable())
//            .username(config.getUsername())
//            .password(config.getPassword())
//            .deserializer(new JsonDebeziumDeserializationSchema())
//            .startupOptions(StartupOptions.initial())
//            .serverId("5400-5440")
//            .jdbcProperties(jdbcProperties)  // 直接传入Properties对象
//            .build();
//            
//        // 3. Kafka Sink
//        Properties kafkaProperties = new Properties();
//        kafkaProperties.setProperty("transaction.timeout.ms", "300000");
//        kafkaProperties.setProperty("request.timeout.ms", "120000");
//        kafkaProperties.setProperty("retries", "3");
//
//        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
//            .setBootstrapServers(config.getBootstrapServers())
//            .setRecordSerializer(KafkaRecordSerializationSchema.builder()
//                .setTopic(config.getTopic())
//                .setValueSerializationSchema(new SimpleStringSchema())
//                .build()
//            )
//            .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
//            .setKafkaProducerConfig(kafkaProperties)  // 使用setKafkaProducerConfig设置属性
//            .build();
//        
//        // 4. 数据处理
//        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC")
//           .map(value -> formatDebeziumLog(value))
//           .filter(Objects::nonNull)
//           .name("Data Processing")
//           .sinkTo(kafkaSink);
//           
//        // 5. 执行
//        env.execute("MySQL CDC to Kafka Job");
//    }
//    
//    /**
//     * 格式化 Debezium 日志
//     * @param log 原始 JSON 日志字符串
//     * @return 格式化后的日志字符串
//     */
//    public static String formatDebeziumLog(String log) {
//        // 解析原始日志为 JSONObject
//        JSONObject rawJson = JSONUtil.parseObj(log);
//
//        // 获取操作类型
//        String op = rawJson.getByPath("op", String.class);
//        if (StrUtil.isBlank(op)) {
//            return null;
//        }
//
//        String operationType;
//        switch (op) {
//	        case "r":
//	        	operationType = "read";
//	        	break;
//            case "c":
//                operationType = "insert";
//                break;
//            case "u":
//                operationType = "update";
//                break;
//            case "d":
//                operationType = "delete";
//                break;
//            default:
//                operationType = "none";
//        }
//
//        // 获取表和数据库信息
//        String table = rawJson.getByPath("source.table", String.class);
//        String database = rawJson.getByPath("source.db", String.class);
//
//        // 获取变化前后的数据
//        JSONObject before = rawJson.getByPath("before", JSONObject.class);
//        JSONObject after = rawJson.getByPath("after", JSONObject.class);
//
//        // 获取时间戳
//        Long timestamp = rawJson.getByPath("ts_ms", Long.class);
//
//        // 构造简洁的格式化输出
//        JSONObject formattedLog = new JSONObject();
//        formattedLog.set("type", operationType);
//        formattedLog.set("database", database);
//        formattedLog.set("table", table);
//        formattedLog.set("time", timestamp);
//        formattedLog.set("before", before);
//        formattedLog.set("after", after);
//
//        return formattedLog.toString();
//    }
}
